﻿using System;
using System.Collections.Generic;
using System.ServiceModel;
using System.Threading;
using Needle.Bus.Engine.Utils;
using Needle.Bus.Messages;

namespace Needle.Bus.Engine.Queues
{
    public class NullSink : EngineComponent, ISink
    {
        public NullSink(string id )
            : this(new NamedMutex(id), id) { }

        public NullSink(object mutex, string id)
            : base(mutex, id)
        { }

        public void Enqueue(IMessage message)
        {
            if (!AcceptMessage(message))
                throw new MessageRejectedException();
        }

        public void Enqueue(IList<IMessage> messages)
        {
            if (!AcceptMessages(messages))
                throw new MessageRejectedException();
        }

        public bool TryEnqueue(IMessage message)
        {
            return AcceptMessage(message);
        }

        public bool TryEnqueue(IList<IMessage> messages)
        {
            return AcceptMessages(messages);
        }

        public IDisposable PrepareForEnqueue(IList<IMessage> messages)
        {
            return new SinkTransaction(this, messages);
        }

        public void Commit(object transactionToken)
        {
            var transaction = transactionToken as SinkTransaction;
            if (transaction == null || transaction.Sink != this)
                throw new ArgumentException("Invalid token", "transactionToken");
            transaction.Commit();
        }

        public void Abort(object transactionToken)
        {
            var transaction = transactionToken as SinkTransaction;
            if (transaction == null || transaction.Sink != this)
                throw new ArgumentException("Invalid token", "transactionToken");
            transaction.Abort();
        }

        public int Count
        {
            get { return 0; }
        }

        private Predicate<IMessage> _filter;
        public Predicate<IMessage> Filter
        {
            get
            {
                ThrowIfDisposed();
                return _filter;
            }
            set
            {
                ThrowIfDisposed();
                var filter = _filter;
                while (Interlocked.CompareExchange(ref _filter, value, filter) != filter)
                    filter = _filter;
            }
        }

        private bool AcceptMessage(IMessage message)
        {
            if (message == null
                || IsDisposed
                || State != CommunicationState.Opened)
                return false;

            var filter = _filter;
            return (filter == null || filter(message)) && !IsDisposed;
        }

        private bool AcceptMessages(IList<IMessage> messages)
        {
            if (messages == null
                || messages.Count == 0
                || IsDisposed
                || State != CommunicationState.Opened)
                return false;

            var filter = _filter;
            if (filter == null)
                return IsDisposed;
            foreach (IMessage message in messages)
            {
                if (filter(message))
                    continue;
                else
                    return false;
            }
            return IsDisposed;
        }

        public override string ToString()
        {
            return String.Format("{0} {1}", GetType().Name, Id);
        }

        protected override void OnOpen(TimeSpan timeout)
        {
            /* do nothing */
        }

        protected override void OnAbort()
        {
            /* do nothing */
        }

        protected override void OnClose(TimeSpan timeout)
        {
            /* do nothing */
        }
    }
}
